import os import utime from machine import UART, Pin #by D@CC #Date: 2021FJun18 #Note: as of today, partially tested # all functions and subroutines are at top of program # 1. bChar() converts 4 char byte into 1 char # 2. serTx_pico() transmits a full serial text message # 3. txParse() parses the outgoing text message and calls txChar # 4. txChar() adds the char to oP which is the output text packet queue # 5. sendThisPack actually transmits a full packet (wo any EOM) # 6. sendLastPack actually transmits text (with at least one EOM) # 7. serRx_pico() receives a full serial text message # testTx transmits many test messages # testRx receives all test messages #calling parms #bChar(strB): #serTx_pico(uartObject,bps=300, txtToBeSent="Hello,World",rxtxEndChar="]",pktSize=16): #txParse(uartObject, thisMsg, bps=300, msgFormat="SMEPX", T5C=-1, pktSize=16, eom="]"): #txChar(char): #sendThisPack(uartObject, oPIn, eom, pktSize): #sendLastPack(oPIn, cntSent, cntUsed, eom = "]", pktSize=16): #serRx_pico(uartObject, bps=300, timeOut=-1, eom="]"): def bChar(strB): #bChar() # given a string like b'z' this returns the "z" # tested by D@CC on 2021FJun15 #strB=input("eg b'h' :") b=strB[2:] #print("b:"+b) c=b[0] #c="z" #print("bChar("+'"'+strB+ '"):'+c) return c #end def def serTx_pico(uartObject,bps=300, txtToBeSent="Hello,World",rxtxEndChar="]",pktSize=16): #serTx_pico# print("txtToBeSent:",txtToBeSent,":") #every parameter has a default setting except the uartObject (which needs to be defined). t5c=50/bps # 50 bit times = 5 character times t10c=100/bps # 100 bit times = 10 character times t1000c=10000/bps # 10000 bit times = 1000 character times # requires a uart object (below) to be created before this function is called #uart = machine.UART(1, 300, parity=None, stop=1, bits=8, tx=Pin(txPin), rx=Pin(rxPin)) print("entering serTx_pico.py") if rxtxEndChar=="]": som = "[" eom = "]" msgOut = """ #S + #M + #E + #P + #C""" #where #A is the letter "A" meaning "16 bit packets (pktSize)" #S is SOM #M is Message text to be sent (will be followed by EOM) #E is EOM #P is the padding characters (enough EOMs to fill the packet) #X is t10c (ten character time delay: "roman numeral X") #C is t100c (100 character time delay: "roman numeral C") #Most messages will have msgOut = "AC" "SMEPX" #Between messages, the transmitter will send "0" every minute indicating that its alive but #has no message to send. #Before sending each message the transmitter will send a single "A" which is 0x41 indicating #that the following message will be in packets of 16 characters. In the future when the #packet size. #changes to 32 characters, the transmitter will precede each message with a single "B" #indicating the each packet will contain 32 characters. This will allow future packet sizes #at least as big as 16x16x4 = 1024 characters and total message lengths of around 32768 without #using any control characters in the "ascii" preamble. # Future versions will have msgOut = "DSNKMEPX" # with packets having the format: # AX SNKMV MV MV etc MEPX #The optional packet of 16 x D permits synchronization # #where #D is a full packet of Delete characters (optional for synchronization #N is the 2-character packet Number #K is the 2-character message character "Kount" including the only the first final EOM #V is t5c (5 character time delay: "roman numeral V") #When a receiver starts up (and between messages) it should start by expecting packets of single #characters. It will read a single 8-bit character expecting it to be a "0" meaning an idle transmitter. #If it is not a "0", it will read another single 8-bit character expecting it to be an "A". If it is not an #"A",the character will be discarded, but the receiver will presume that it is reading a character #stream that is mid-message. It will read single 8-bit characters searching for a "]" which will indicate #that this is the last packet of the message. It will keep reading and discarding all the "]" that it sees. #After the last "]", the #next 8-bit character should be a "0" or an "A". The receiver should keep reading and discarding the "0" #characters. When an "A" is finally received (while reading single 8-bit characters), the receiver knows that #this means subsequent packets will be packets of 16 characters. The receiver is now fully #synchonized with the transmitter. If something other than an "A" is received, the receiver should #begin looking for a "]" ()meaning "end of message" again. If the state of "being in mid-message #continues 10 times, a human operator should be informed the that this ERROR is occuring. # #send the message txParse(uart,txtToBeSent) #xyz return True #end def serTx_pico() def txParse(uartObject, thisMsg, bps=300, msgFormat="SMEPX", T5C=-1, pktSize=16, eom="]"): #txParse() show("thisMsg:",thisMsg) # # uartObject is the uart being used # thisMsg is the full text message to be sent (It should not contain (EOM="]") they will become "?") # (An EOM is added as the last character of a message, do not put the EOM in the msg) # bps is the baud rate (bits/sec) # msgFormat is the form of the Msg e.g. SMEPX (first E is counted as last char of M) # (P is enough additional EOMs to pad to packetSize=16) # (t5c is interval between packets (-1 defaults to t5c) # t5c indicates time interval of 5 characters. The transmitter will follow each packet with # this time delay # packetSize defaults to 16 # eom is the definition of the 'EOM" character (usually "]") #oP is the full outgoing Packet that is sent by sendThisPack # or by sendLastPack oP="" # outgoing msg is initially empty if T5C==-1: # if the calling routine doesn't specify t5c=50/bps t10c=100/bps t1000c=10000/bps else: t5c=T5C #if the calling routine does specify T5C t10c=T5C*2 t1000c=T5C*200 #end if if eom=="]": som="[" #end if # in 2021 msgOut="SMEPX" lenM=len(thisMsg)+1 #length of thisMsg -1 (+1 for the EOM character cntUsed=0 # number of characters in msgOut used so far cntSent=0 # number of character actually sent (incl all chars sent) isEOMsent=False #EOM not sent yet show("msgFormat:",msgFormat) for msgPart in msgFormat: # do each message segment if msgPart=="S": # send SOM print("send SOM") oP=oP+txChar(som) oP=oP+txChar("1") # in future these 2 characters will contain parity or CRC oP=oP+txChar("6") # today they contain the packet size (ie character count) cntSent=cntSent+3 #3 actually sent now print() print("oP:"+oP+":") #end if if msgPart=="X": # time delay doesn't send any characters utime.sleep(t10c) #end if if msgPart=="M": # sends the msg followed by 1 eom print("sending M") isLastPack=True while (lenM-cntUsed) >= 1 : #lenM incl the EOM, cntUsed does also (after the EOM is sent) isLastPack=False # the packet being sent should not include any EOM characters. (EOM is only in the last pk) # send a packet of data w/o eom lastChar=cntUsed+pktSize show("lastChar",lastChar) #thisMsg="12345678901234567" print("thisMsg:"+thisMsg+":") show("cntUsed",cntUsed) show("lastChar",lastChar) show("thisMsg[cntUsed:lastChar]",thisMsg[cntUsed:lastChar]) opToSend=thisMsg[cntUsed:lastChar] print("opToSend:"+opToSend+":") if len(opToSend)>=16 : print() print("before sendThisPack oP:"+oP+":") cntUsedFromMsg=sendThisPack(uartObject,opToSend, eom, pktSize) #returns 16 should print the Pkt oP="" show("cntUsedFromMsg:",cntUsedFromMsg) cntUsed=cntUsed+cntUsedFromMsg # should be 16 ...... to verify cntSent=cntSent+16 show("lenM",lenM) show("cntUsed",cntUsed) isLastPacket=False else: print("need to send padded Packet") isLastPacket=True #end if leftToSend=lenM-cntUsed-1 show("leftToSend:",leftToSend) if isLastPacket==True : # will send the Last Packet (containing at least 1 EOM) print("isLastPack is True") #useStr=thisStr cntUsedFromMsg=len(opToSend) toAdd=16-cntUsedFromMsg show("toAdd:",toAdd) for x in range(toAdd): # must insert correct range !!!!! oP=oP+txChar(eom) # must insert correct char from fullMsg !!!!! cntSent=cntSent+1 #end for print() print("before sendThisPack oP:"+oP+":") cntSent=sendThisPack(uartObject,oP, eom, pktSize) #returns 16 should print the Pkt oP="" cntUsed=cntUsed+cntUsedFromMsg show("cntUsed:",cntUsed) #end if print("after Last Packet of Msg") toSend=lenM-cntUsed-1 show("toSend:",toSend) print("end of while") #end while # the above sent all packets except the last one print("after end of while") #end if "M" print("after while") if False : if msgPart=="P": # P means Padding (with more EOMs) if any needs to be sent print("msgPart=P") if False : if True: # to be decided oP=oP+txChar(eom) # how many eoms of padding !!!!! cntUsed=cntUsed+1 cntSent=cntSent+1 #end if show("cntUsed",cntUsed) strSent=thisMsg[cntUsed:cntUsed+16] show("strSent:",strSent) cntUsedFromMsg = sendLastPack(uartObject, strSent, eom, pktSize) # should print the Pkt cntUsed=cntUsed+cntUsedFromMsg cntSent=cntSent+pktSize #pktSize is usually 16 #end if #end if "P" #end if False #end for # all msg parts have now been sent # should check that cntUsed = len(Msg) +1 # should check that cntSent is a muliple of pktSize(eg 16) return #from parsing and transmitting #end def def txChar(char): #txChar() #simply records that "char" has been sent # just prints it during testing # doesn't actually send the char print(char, end='') #"end" suppresses the line feed return char #end def def sendThisPack(uartObject, opIn, eom, pktSize): #sendThisPack() # #actually sends the 16 characters in oP print() print("in sendThisPack") oP=opIn lenopIn=len(opIn) if (lenopIn==pktSize)==False: print("ERROR",pktSize, "is not the size of the packet to send:",lenopIn) #end if pos=opIn.find(eom) if pos==-1: # -1 if not found pass else: print("WARNING E in serTx", eom, "found in message") oP=oP.replace(eom,"?") #end if uartObject.write(oP) print("tx sent:"+oP+":") #keep user informed during testing return pktSize #end def def mpyRead(uart): #keep read1 character packet until timeOut txtRead="" bRead="" while uart.any(): aByte=str(uart.read(1)) #show("aByte:",aByte) bRead=bChar(aByte) #show("bRead:",bRead) txtRead=txtRead+bRead #end while return txtRead #end def def sendLastPack(uartObject, opIn, eom = "]", pktSize=16): #sendLastPack() #actually sends the character packet in oP # but pads the packet with "eom"s first # replace any eom with "?" lenopIn=len(opIn) oP=opIn pos=oP.find(eom) if pos==-1: # Good: eom not found pass else: print("ERROR E in serTx", eom, "found in message") oP.replace(eom,"?") #end if len=len(oP) #pad packet with eoms show("lenOp",lenOp) for k in range(16,lenOp): oPtoSend=oPtoSend+eom # add eoms for padding #end for lenOp=len(oP) # is the pktSize correct if (lenOp==pktSize)==False: print("ERROR",pktSize, "is not the size of the packet to send:",lenOp) #end if #send the 16 character packet uartObject.write(oPtoSend) print("tx padded (if needed) and sent:"+oP+":") return lenOpIn # return # of chars used. #end def # serRx_pico() receives a full serial text message #serRx_pico() def serRx_pico(uartObject, bps=600, timeOut=-1, eom="]"): #uart = machine.UART(1,bps, parity=None, stop=1, bits=8, rx=Pin(5), tx=Pin(4)) print("my uart:",uart) #print("timeOut is set to:",timeOut) if timeOut==-1: t100c = 1000/bps timeOutUse=int(t100c+1.0) #at least 1 sec #show("timeOutUse:",timeOutUse) else: timeOutUse=3 #in seconds #end if #timeOutUse=t100c time=utime.time() timeNow=time #show("timeNow:",timeNow) timeFail=timeNow+timeOutUse # defaults to a 3 second TimeOut #print("waiting my:",timeOutUse," seconds for data receipt. . . . .") oP="" # reset input buffer #show("time:",time) #show("timeFail:",timeFail) while time < timeFail : #keep reading any 1 character packets until timeOut #print("any") strA=mpyRead(uart) #show("strA:",strA) oP=oP+strA #show("oP:",oP) #end if time=utime.time() #end while #show("oP:",oP) lenOp=len(oP) if oP=="": print("ERROR T in serRx (): timeOut after",timeOutUse,"seconds") lenOp=len(oP) #end if if lenOp > 0: if oP.find("?")==-1: pass # ? not found else: print("WARNING '?' seen in message received.") #end if #end if #print("serRx() received:"+oP+":") #during testing return oP #end def serRx_pico(uart,300,timeOut(in secs)) def show(strA,parmA): # show print(strA,parmA) return #end def ########################################################################################## progName="testTxC_pico.py" print("progName:"+progName) if True: bps=600 txPin=5 rxPin=4 # #print sys info print(os.uname()) #os, utime and machine were imported at top #indicate program started visually (by blinking the green LED) led_onboard = machine.Pin(25, machine.Pin.OUT) led_onboard.value(0) # onboard LED OFF for 0.5 sec utime.sleep(0.5) led_onboard.value(1) # leaving it ON #create uart object sample code if False : #self.uart = UART(1, 115200, parity=None, stop=1, bits=8, rx=Pin(rxPin), tx=Pin(txPin)) pass #end if #create actual uart object uart = machine.UART(1,bps, parity=None, stop=1, bits=8, rx=Pin(5), tx=Pin(4)) #print uart info #uart = machine.UART(1) print(uart) # during testing # create uartObject #uart = machine.UART(1,bps, parity=None, stop=1, bits=8, tx=txPin, rx=rxPin) # send a single "A" which is 0x41 ("A"-0x40 * 16 is the packet length which is 16 # initialization is complete uart.write("A") #every group of messages is preceeded by a single "A" # prior to the "A", the transmitter should send a synchronizing sequence to permit # the (human) receiver to "guess" or "determine" the baud rate (the bps) # hopefully the operator of the transmitter and receiver is the same human. # the serRx_pico should be running in a separate thread. #def serTx_pico(uartObject,bps, txtToBeSent="[UUUUUUUU]",rxtxEndChar="]"): serTx_pico(uart,bps, txtToBeSent="Hello,JoeMaxRoyBob") if False: serTx_pico(uart,bps, txtToBeSent="H") serTx_pico(uart,bps, txtToBeSent="H") serTx_pico(uart,bps, txtToBeSent="H]") serTx_pico(uart,bps, txtToBeSent="[H]") serTx_pico(uart,bps, txtToBeSent="2.97") serTx_pico(uart,bps, txtToBeSent="99.987654") serTx_pico(uart,bps, txtToBeSent="ADC0:(0x4000,1622428309.000j)") serTx_pico(uart,bps, txtToBeSent="1234567890123") serTx_pico(uart,bps, txtToBeSent="12345678901234") serTx_pico(uart,bps, txtToBeSent="Hello,World") serTx_pico(uart,bps, txtToBeSent="1234567890123456789012345678901234567890") #end if if False: # the above is the current format of the rxtx protocol # The som,eom,t10c and are 1-character special codes # each must be converted into outgoing text or actions or both # all normal text does not need to be converted before transmission # # The txParse routine analyses the msgOut Format and separates it into # special codes (text and actions). Each special code is sent to the txChar() # routine to be processed. Each character of outgoing text is sent to the # txChar() routine to be processed. # # The serTx_pico() routine is used to # transmit a special code of text ("S" and "M") # cause a time delay (V is "t5c X is "t10c " and M is "t1000c") # transmit msg as 16-byte packets (msg should not contain "]" which is replaced by "?" ) # # early testing Format is "SMEPX" # # Note that the "P" also inserts at least 1 "]" character and # enough extra "]" characters (or none) to make a message with an exact multiple # of 16 characters. This is because there are always 16 characters sent per # packet. Each packet is followed by a time delay of t5c. # The last packet is always followed by a time delay of t10c. # These time delays permit the receiver to resync with the transmitter. # These time delays are never included in the character count of the # message # The serRx_pico() routine times out after an interval of 1000 character times: t1000c pass #end if # This should be run this as a separate thread # print whatever has been received bps=600 t100c=1000/bps utime.sleep(t100c) #wait for an interval of 100 char times #end if #uart = machine.UART(1,bps, parity=None, stop=1, bits=8, tx=Pin(txPin), rx=Pin(rxPin)) bps=600 uart = machine.UART(1,bps, parity=None, stop=1, bits=8, tx=Pin(4), rx=Pin(5)) print("uart:",uart) if False: uart.write("hello") t100c=1000/bps utime.sleep(t100c) #wait for an interval of 100 char times uart.write("Hello,World") utime.sleep(t100c) #wait for an interval of 100 char times utime.sleep(1) #1 sec #if end if False: print("receiving mpy routine") while uart.any(): print(uart.read(1)) #end while #end if charIn="" if False: print("receiving mpy routine") if uart.any() : #print("reading now") charIn=uart.read(1) #show("charIn:",charIn) #oP=oP+uartObject.read(1) #end if print("end of mpy routine") #end if show("charIn:",charIn) #end if print("receiving my routine") msgIn=serRx_pico(uart,bps=600) #times out after t100c seconds (100 char-times) print("msgIn:",msgIn) print("end of serRX()") #end /testTxC_pico.py